
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sstream>
#include <glib.h>
#include <assert.h>
#include <libmemcached/memcached.h>
#include "configParser.h"
#include "define.h"
#include "Debug.h"
#include "Helper.h"
 
using namespace std;
Helper * Helper::_instance = NULL;

Helper::Helper() 
{
	m_sConfFile = "";
        m_sets = 0;
        m_gets = 0;
        m_pEvents = NULL;
}

Helper::~Helper()
{
    if(m_pEvents != NULL)
       delete m_pEvents;
}

Helper * Helper::Instance(const char *confFile)
{
	if (NULL == _instance) 
	{
		_instance = new Helper();
		if ( !_instance->Startup(confFile) ) {
			delete _instance; 
			_instance = NULL;
		}
	}
	return _instance;
}

void Helper::Release(bool bForce)
{
	if (_instance != NULL) 
	{
		if(bForce)
		{
			delete _instance; 
			_instance = NULL;
		}
	}
}

bool  Helper::Startup(const char *confFile)
{
	if(confFile != NULL)
		m_sConfFile  = confFile;
	if(m_sConfFile.length() < 1)
	{
		ERRO<<": no helper.conf be given!"<<endl;
		return false;
	}
	struct stat st;
	if (lstat(confFile, &st) != 0)
	{
		ERRO<<":"<<confFile<<" is not exist!"<<endl;
		return false;
	}

	configParser cfg;
	map <string, string > g_Confs;
	g_Confs = cfg.Parser(confFile);
	if(g_Confs.size() <= 0)
	{
		cerr<<"error to explain conf :"<< confFile <<endl;
		return false;
	}

	map <string , string >::iterator iter;
	for(iter=g_Confs.begin(); iter != g_Confs.end(); iter++)
	{
		cout<< iter->first.c_str()<<"="<<iter->second.c_str()<<endl;
	}



	int iGCSize = 100000;
	if( g_Confs.find("gcSize") != g_Confs.end())
	{
		string stype  = g_Confs["gcSize"];
		iGCSize = atoi(stype.c_str());
	}

	float fGCfactor = 0.6;
	if( g_Confs.find("gcFactor") != g_Confs.end())
	{
		string stype  = g_Confs["gcFactor"];
		fGCfactor = atof( stype.c_str() );
		if(fGCfactor <= 0 || fGCfactor >= 1 )
			fGCfactor = 0.6;
	}

	int kickout = 84000;
	if( g_Confs.find("kickout") != g_Confs.end())
	{
		string stype  = g_Confs["kickout"];
		kickout = atoi(stype.c_str());
		if(kickout < 1  )
			kickout =  1;
	}

	string scName;
	if( g_Confs.find("consumerName") != g_Confs.end())
	{
		scName  = g_Confs["consumerName"];
	}

	string scLibrary;
	if( g_Confs.find("consumerLib") != g_Confs.end())
	{
		scLibrary  = g_Confs["consumerLib"];
		struct stat st;
		if (lstat((char *)scLibrary.c_str() , &st) != 0)
		{
			ERRO<< " file not exist:"<< scLibrary <<endl ;
			scLibrary = ""; 
		}
	}
	string scConfFile;
	if( g_Confs.find("consumerConf") != g_Confs.end())
	{
		scConfFile  = g_Confs["consumerConf"];
		struct stat st;
		if (lstat((char *)scConfFile.c_str() , &st) != 0)
		{
			ERRO<< " file not exist:"<< scConfFile <<endl ;
			scConfFile  = ""; 
		}
	}
	if( m_pEvents == NULL)
	{
		m_pEvents  = new Events(scName.c_str(), scLibrary.c_str(), scConfFile.c_str(), iGCSize, fGCfactor, kickout);
		if(!m_pEvents->Initialize())
		{
			ERRO<<" Events can't be Initialize"<<endl;
			delete m_pEvents;
			m_pEvents = NULL;
			return false;
		}
	}
	else
	{
		ERRO<<"Events is not NULL ??? !!!"<<endl;
		return false;
	}
	if( g_Confs.find("returnType") != g_Confs.end())
	{
		string itype  = g_Confs["returnType"];
		m_pEvents->setReturnType(atoi(itype.c_str()));
	}

	if( g_Confs.find("updateMax") != g_Confs.end())
	{
		string stype  = g_Confs["updateMax"];
		int imax = atoi(stype.c_str());
		DEBUG2("set updateMax = ", imax);
		m_pEvents->setUpdateMax(imax);
	}

	if( g_Confs.find("returnMaxCounts") != g_Confs.end())
	{
		string stype  = g_Confs["returnMaxCounts"];
		int imax  = atoi(stype.c_str());
		DEBUG2("set returnMaxCounts = ", imax);
		m_pEvents->setReturnMaxCounts(imax);
	}

	if( g_Confs.find("update_port") != g_Confs.end())
	{
		string stype  = g_Confs["update_port"];
		m_update_port  = atoi(stype.c_str());
		DEBUG2("set update_port  = ", m_update_port);
	}

	return true;
}

int Helper :: Process(map < int, string > &keys, conn *c)
{
	m_gets++;
	int iSize = keys.size();
	if(iSize <= 0)
		return -1;
	int CMD = 1;
	string cmd = keys[CMD];
	int iret = BRUNCE_ERR_SYSTEM ;
	if(cmd.compare("friends") == 0 && iSize == 5 )
	{ //friends +  user + limit + offset + END
		iret = Get_Friends((char *)keys[2].c_str(), atoi(keys[3].c_str()), atoi(keys[4].c_str()) , c);
	}
	else if(cmd.compare("tfriends") == 0 && iSize == 6 )
	{ //tfriends + user + type + limit + offset + END
		iret = Get_Type_Friends((char *)keys[2].c_str(), atoi(keys[3].c_str()), atoi(keys[4].c_str()), atoi(keys[5].c_str()) , c);

	}
	else if(cmd.compare("relation") == 0 && iSize == 6)
	{ //relation + usera + userb + limit + offset + END
		iret = Get_Relations((char *)keys[2].c_str(),(char *)keys[3].c_str(),  atoi(keys[4].c_str()), atoi(keys[5].c_str()) , c);
	}
        else if(cmd.compare("stats") == 0 && iSize == 4)
        { // stats + var + val + END
		ostringstream os;
		if(keys[2].compare("updatemax") == 0)
			os<< m_pEvents->setUpdateMax( atoi(keys[3].c_str()));
		if(keys[2].compare("returntype") == 0)
			os<< m_pEvents->setReturnType(atoi(keys[3].c_str()));
		if(keys[2].compare("returnmax") == 0)
			os<< m_pEvents->setReturnMaxCounts(atoi(keys[3].c_str()));
		if(keys[2].compare("factor") == 0)
			os<< m_pEvents->setFactor( atof(keys[3].c_str()));
		if(keys[2].compare("kickout") == 0)
			os<< m_pEvents->setKickout( atoi(keys[3].c_str()));
		if(keys[2].compare("gcsize") == 0)
			os<< m_pEvents->setGCSize(atoi(keys[3].c_str()));
                if(os.str().length())
                {
                    bcopy(os.str().c_str(), c->wcurr, os.str().length());
                    c->wbytes += os.str().length();
                    iret = BRUNCE_SUCCESS;
                } 
        }
        else if(cmd.compare("reset") == 0 && iSize >= 3)
	{ // reset + user + [remote ? true :false] END
                m_sets++;
                bool remote = false;
                if(iSize >=4)
                {
                     int ir = atoi(keys[3].c_str());
                     ir > 0 ? remote = true : remote = false;
                }  
                ReBuild(keys[2].c_str(), remote);
		bcopy("OK", c->wcurr, 2);
		c->wbytes += 2;
	}
	return iret;
}

int Helper :: Process(const char *cmd, conn *c) // char **wbuf, int &wsize, int &wbytes)
{
	m_gets++;
	DEBUG2("key = ", cmd);
	if(strcmp("__stat__", cmd) == 0)
	{
		c->wbytes +=  sprintf(c->wcurr, "imok");
		return BRUNCE_SUCCESS;
	}
	else
		return Get_Infomation((char *)cmd, c);
}

int Helper :: Timer(long lt, int gap)
{
/*	if(m_pEvents != NULL)
		m_pEvents->Process_Worklist(lt);
*/
	return 0;
}

int Helper:: Get_Relations(char *usera, char *userb, int limit, int offset,  conn *c)
{
	assert(m_pEvents != NULL && usera != NULL && userb != NULL);
        return  m_pEvents->Get_Users_Graph(usera, userb, limit, offset, c);
}

int Helper:: Get_Friends(char *user, int limit, int offset, conn *c) // char **wbuf, int &wsize, int &wbytes)
{ // ulen +uval 
        assert(user != NULL && m_pEvents != NULL);
	return  m_pEvents->Get_User_Friends(user, limit, offset, c); // wbuf, wsize, wbytes);
}

int Helper:: Get_Type_Friends(char *user, int type, int limit, int offset, conn *c) // char **wbuf, int &wsize, int &wbytes)
{ 
	return m_pEvents->Get_User_Friends(user, type, limit, offset, c); // wbuf, wsize, wbytes);
}

int Helper:: Get_Infomation(char *user, conn *c)
{ 
	assert(m_pEvents != NULL && user != NULL );
	return  m_pEvents->Get_Users_Value(user, c); // wbuf, wsize, wbytes);
}


char * Helper :: Stats(char *ext_stats)
{
        assert(m_pEvents != NULL);
        string status;
        m_pEvents->Get_Status(status);
        char hostname[128];
        gethostname(hostname, sizeof(hostname)) ;
        status.append("STAT server ");
        status.append(hostname);
        status.append("\r\nSTAT gets ");
        int len = sprintf(hostname, "%ld\r\n", m_gets);
        status.append(hostname, len);
        status.append("STAT sets ");
        len = sprintf(hostname, "%ld\r\n", m_sets);
        status.append(hostname, len);
        if(ext_stats != NULL)
                status.append(ext_stats);

	ostringstream os;
        os<<"STAT updata_max "<< m_pEvents->setUpdateMax(0)<<"\r\n";
        os<<"STAT return_type "<< m_pEvents->setReturnType(-1)<<"\r\n";
        os<<"STAT return_max "<< m_pEvents->setReturnMaxCounts(0)<<"\r\n";
        os<<"STAT gc_factor "<< m_pEvents->setFactor(0)<<"\r\n";
        os<<"STAT gc_kickout "<< m_pEvents->setKickout(0)<<"\r\n";
        os<<"STAT gc_size "<< m_pEvents->setGCSize(0)<<"\r\n";
        status.append(os.str());
        return (char *)status.c_str();
}

bool Helper:: Clear()
{ // ulen +uval + ulen + uval
	assert(m_pEvents != NULL );
	return	m_pEvents->Clear();
}

int Helper:: Update(const char *key, int klen, int flags, time_t expire, const char *data, int dlen)
{
	m_sets++;
	if(strcmp(key,"CONFIGURE") == 0)
	{
             /*updatemax=a;returntype=3;returnmax=8;factor=0.3;kickout=333;gcsize=20  */
                map < string, string > cfg; 
		char **columns = g_strsplit(data , ";", -1);
		int columns_num =  g_strv_length(columns);
		for(int i=0; i< columns_num ; i++ )
		{ // g_strstrip(columns[i]);
			string stemp = g_strstrip(columns[i]);
			size_t idx = stemp.find('=', 0);
			if(idx>0 && idx < stemp.length())
			{
				string key = stemp.substr(0, idx);
				string val = stemp.substr(idx + 1, stemp.length()- idx -1);
				cfg[key] = val;
			}
		}
		g_strfreev(columns);
		if( cfg.find("gcsize") != cfg.end())
		{
			string s  = cfg["gcsize"];
                        m_pEvents->setGCSize( atoi(s.c_str()));
		}
		if( cfg.find("updatemax") != cfg.end())
		{
			string s  = cfg["updatemax"];
                        m_pEvents->setUpdateMax( atoi(s.c_str()));
		}
		if( cfg.find("returntype") != cfg.end())
		{
			string s  = cfg["returntype"];
                        m_pEvents->setReturnType( atoi(s.c_str()));
		}
		if( cfg.find("returnmax") != cfg.end())
		{
			string s  = cfg["returnmax"];
                        m_pEvents->setReturnMaxCounts( atoi(s.c_str()));
		}
               	if( cfg.find("factor") != cfg.end())
		{
			string s  = cfg["factor"];
                        m_pEvents->setFactor( atof(s.c_str()));
		}
		if( cfg.find("kickout") != cfg.end())
		{
			string s  = cfg["gcsize"];
                        m_pEvents->setKickout( atoi(s.c_str()));
		}
	}
	return 0;
}

int Helper :: Check( void *args )
{
	int *port = (int *)args;
	DEBUG1(" begin to Check!");
	memcached_st * memc = memcached_create(NULL);
	memcached_return rc = MEMCACHED_SUCCESS;
	rc = memcached_server_add(memc,(char *)"127.0.0.1",(unsigned int)(*port));
	if (rc != MEMCACHED_SUCCESS){
		perror("Connect to memcache server");
		memcached_free( memc );
		return 0;
	}
	memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NO_BLOCK, 1);
	memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_CONNECT_TIMEOUT, 1);
	memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 0);
	size_t vlen;
	uint32_t flag;
	char *retval = memcached_get(memc,"__stat__", 8, &vlen, &flag, &rc);
	if (rc == MEMCACHED_SUCCESS){
		if (retval != NULL)
			free(retval);
	}
	if (rc != MEMCACHED_SUCCESS)
	{
		printf("Failure to communicate with servers (%s)\n", memcached_strerror(memc, rc));
		for(int i =0; i< 3; i++)
		{
			usleep(100);
			retval = memcached_get(memc, "__stat__", 8, &vlen, &flag, &rc);
			if(rc == MEMCACHED_SUCCESS){
				free(retval);
				break;
			}
		}
		if(rc != MEMCACHED_SUCCESS){
			memcached_free(memc);
			exit(1);
			//abort();
		}
	}
	memcached_free( memc );
	DEBUG1("end Check!");
	return 0;
}

int Helper :: ReBuild( const char *usr, bool remote )

{
        if( usr == NULL || strlen(usr) < 1 )
               return 0;

	if(remote)
	{
		if( m_update_port < 1)
			return 0;
		memcached_st * memc = memcached_create(NULL);
		memcached_return rc = MEMCACHED_SUCCESS;
		rc = memcached_server_add(memc,(char *)"127.0.0.1",(unsigned int)(m_update_port));
		if (rc != MEMCACHED_SUCCESS){
			perror("Connect to memcache server");
			memcached_free( memc );
			return 0;
		}
		memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NO_BLOCK, 1);
		memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_CONNECT_TIMEOUT, 1);
		memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 0);
		size_t vlen;
		uint32_t flag;
		char *retval = memcached_get(memc, usr, strlen(usr), &vlen, &flag, &rc);
		if (rc == MEMCACHED_SUCCESS){
			if (retval != NULL)
				free(retval);
		}
		else
		{
			printf("Failure to communicate with servers (%s)\n", memcached_strerror(memc, rc));
			retval = memcached_get(memc, usr, strlen(usr) , &vlen, &flag, &rc);
			if(rc == MEMCACHED_SUCCESS)
				free(retval);
		}
		memcached_free( memc );
                return 0;
	}
        else
              m_pEvents->Reload_User(usr);
	return 0;
}


